package com.esfak47.mediafetcher.app.service.impl;

import com.esfak47.common.PageResult;
import com.esfak47.common.job.Job;
import com.esfak47.common.job.JobManager;
import com.esfak47.common.lang.Assert;
import com.esfak47.common.lang.Inject;
import com.esfak47.common.logger.Logger;
import com.esfak47.common.logger.LoggerFactory;
import com.esfak47.mediafetcher.core.event.Event;
import com.esfak47.mediafetcher.core.event.EventGroupEnum;
import com.esfak47.mediafetcher.core.event.EventSender;
import com.esfak47.mediafetcher.core.utils.JobUtils;

import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executor;
import java.util.function.BiConsumer;

/**
 * @author tony
 */
public class DefaultJobManagerImpl implements JobManager {

    private Logger logger = LoggerFactory.getLogger(DefaultJobManagerImpl.class);
    private EventSender eventSender;

    private Map<String, Job> jobContainer;

    private Map<String, CompletableFuture> futureMap;

    private Executor jobExecutor;

    public DefaultJobManagerImpl() {}

    private static <T> List<T> getPartOf(Collection<T> collection, int start, int end) {
        Assert.notNull(collection, "collection should not be null");

        int size = collection.size();
        if (size <= start) {
            return Collections.emptyList();
        } else if (size > start && size < end + 1) {
            int i = 0;
            List<T> list = new ArrayList<>();
            Iterator<T> iterator = collection.iterator();
            while (iterator.hasNext() && i < start) {
                i++;
                iterator.next();
            }
            while (iterator.hasNext()) {
                list.add(iterator.next());
            }
            return list;

        } else {
            int i = 0;
            List<T> list = new ArrayList<>();
            Iterator<T> iterator = collection.iterator();
            while (iterator.hasNext() && i < start) {
                i++;
                iterator.next();
            }
            while (iterator.hasNext() && i <= end) {
                list.add(iterator.next());
            }
            return list;
        }

    }

    @Override
    public boolean submit(Job job) {
        Assert.notNull(job, "job should not be null");
        String jobId = job.getId();
        Assert.notNull(jobId, "jobId should not be null");
        if (jobContainer.containsKey(jobId) || futureMap.containsKey(jobId)) {
            return false;
        }
        jobContainer.put(jobId, job);
        JobUtils.sendSubmitEvent(eventSender, job);
        CompletableFuture<Void> voidCompletableFuture = CompletableFuture.runAsync(job, jobExecutor);
        if (!job.isAsync()) {
            futureMap.put(jobId, voidCompletableFuture);
            voidCompletableFuture.whenComplete(new BiConsumer<Void, Throwable>() {
                @Override
                public void accept(Void aVoid, Throwable throwable) {
                    job.setFinish(voidCompletableFuture.isDone());
                    if (job.isFinish()) {
                        job.setProcess(100);
                    }
                    if (throwable != null) {logger.error("job " + jobId + " execute failed", throwable); }
                    job.setExitUnexpectedly(voidCompletableFuture.isCompletedExceptionally());
                    job.setCancelled(voidCompletableFuture.isCancelled());
                    job.setFinishTime();
                    futureMap.remove(jobId);
                }
            });
        }
        eventSender.send(new Event(EventGroupEnum.JOB, "submit", job));
        return true;
    }

    @Override
    public boolean cancel(String jobId) {
        Assert.notNull(jobId, "jobId should not be null");
        if (!jobContainer.containsKey(jobId) || !futureMap.containsKey(jobId)) {
            return false;
        }
        JobUtils.sendCancelEvent(eventSender, jobContainer.get(jobId));
        CompletableFuture completableFuture = futureMap.get(jobId);
        return completableFuture.cancel(true);
    }

    @Override
    public boolean delete(String jobId) {
        Assert.notNull(jobId, "jobId should not be null");
        if (!jobContainer.containsKey(jobId) || !futureMap.containsKey(jobId)) {
            return true;
        }
        boolean cancel = this.cancel(jobId);

        if (cancel) {
            final Job remove = jobContainer.remove(jobId);
            JobUtils.sendDeleteEvent(eventSender, remove);
            return true;
        }

        return false;
    }

    @Override
    public PageResult<Job> getJobs(int page, int pageSize) {

        int start = (page - 1) * pageSize;
        int end = page * pageSize - 1;

        Collection<Job> values = jobContainer.values();

        return PageResult.createPageResult(page, pageSize, values.size(), getPartOf(values, start, end));

    }

    @Override
    public Job getJob(String id) {
        return jobContainer.get(id);
    }

    @Inject
    public void setJobContainer(Map<String, Job> jobContainer) {
        this.jobContainer = jobContainer;
    }

    @Inject
    public void setFutureMap(Map<String, CompletableFuture> futureMap) {
        this.futureMap = futureMap;
    }

    @Inject
    public void setJobExecutor(Executor jobExecutor) {
        this.jobExecutor = jobExecutor;
    }

    @Inject
    public void setEventSender(EventSender eventSender) {
        this.eventSender = eventSender;
    }
}